//---------------------------------------------------------------------------
// Process Class
// Author : Woo-Hyun Kim
// Email : woorung@icu.ac.kr
// Date : Jul. 7, 2006
//---------------------------------------------------------------------------
#include "CProcess.h"
//---------------------------------------------------------------------------
CProcess::CProcess()
{
	status = -1;

	worker_pid = -1;
	pid = -1;
	ppid = -1;
	cpid[CHILD_PROC_1] = -1; // 1st child process before run
	cpid[CHILD_PROC_2] = -1; // 2nd child process after run

	// for synchronization
	master_sync = NULL;
	filter_sync = NULL;
	worker_sync = NULL;

	// child to parent
	up_pipefd[0] = -1;
	up_pipefd[1] = -1;

	// parent to child
	down_pipefd[0] = -1;
	down_pipefd[1] = -1;

	// child to parent
	err_pipefd[0] = -1;
	err_pipefd[1] = -1;

	verbose = 0;
	err_log = NULL;
}
//--------------------------------------------------------------------------
CProcess::CProcess(int verbose)
{
	status = -1;

	worker_pid = -1;
	pid = -1;
	ppid = -1;
	cpid[CHILD_PROC_1] = -1; // 1st child process before run
	cpid[CHILD_PROC_2] = -1; // 2nd child process after run

	// for synchronization
	master_sync = NULL;
	filter_sync = NULL;
	worker_sync = NULL;
	
	// child to parent
	up_pipefd[0] = -1;
	up_pipefd[1] = -1;

	// parent to child
	down_pipefd[0] = -1;
	down_pipefd[1] = -1;

	// child to parent
	err_pipefd[0] = -1;
	err_pipefd[1] = -1;

	this->verbose = verbose;
	err_log = NULL;
}
//--------------------------------------------------------------------------
CProcess::~CProcess()
{
	if(master_sync!=NULL) delete master_sync;
	if(filter_sync!=NULL) delete filter_sync;
	if(worker_sync!=NULL) delete worker_sync;

	if(up_pipefd[0]>0) close(up_pipefd[0]);
	if(up_pipefd[1]>0) close(up_pipefd[1]);

	if(down_pipefd[0]>0) close(down_pipefd[0]);
	if(down_pipefd[1]>0) close(down_pipefd[1]);

	if(err_pipefd[0]>0) close(err_pipefd[0]);
	if(err_pipefd[1]>0) close(err_pipefd[1]);

	if(err_log!=NULL) delete err_log;
}
//--------------------------------------------------------------------------
void CProcess::before_run()
{
	// override the user-defined code
}
//--------------------------------------------------------------------------
void CProcess::run()
{
	// override the user-defined code
}
//--------------------------------------------------------------------------
void CProcess::after_run()
{
	// override the user-defined code
}
//--------------------------------------------------------------------------
pid_t CProcess::get_wpid()
{
	return worker_pid;
}
//--------------------------------------------------------------------------
pid_t CProcess::get_pid()
{
	return pid;
}
//--------------------------------------------------------------------------
// only valid in parent process
//--------------------------------------------------------------------------
pid_t CProcess::get_cpid(int i)
{
	// only valid in parent process
	return cpid[i];
}
//--------------------------------------------------------------------------
pid_t CProcess::get_ppid()
{
	return ppid;
}
//--------------------------------------------------------------------------
int CProcess::get_status()
{
	return status;
}
//--------------------------------------------------------------------------
// child write to parent
//--------------------------------------------------------------------------
size_t CProcess::write_upstream(const void *buf, size_t size)
{
	return write(up_pipefd[1], buf, size);
}
//--------------------------------------------------------------------------
// child write to parent
//--------------------------------------------------------------------------
size_t CProcess::write_upstream(const char *buf)
{
	size_t size = strlen(buf);

	return write(up_pipefd[1], buf, size);
}
//--------------------------------------------------------------------------
// parent read from child 
// # Attention : it must be overridden in subclasses
//--------------------------------------------------------------------------
int CProcess::read_upstream(const char *buf)
{
	cout << buf;

	return 1;
}
//--------------------------------------------------------------------------
int CProcess::read_upstream(const void *buf, size_t size)
{
	for(size_t i=0;i<size;i++) cout << *((char *)buf + i);

	return 1;
}
//--------------------------------------------------------------------------
// parent write child 
//--------------------------------------------------------------------------
size_t CProcess::write_downstream(const void *buf, size_t size)
{
	return write(down_pipefd[1], buf, size);
}
//--------------------------------------------------------------------------
// parent write child 
//--------------------------------------------------------------------------
size_t CProcess::write_downstream(const char *buf)
{
	size_t size = strlen(buf);

	return write(down_pipefd[1], buf, size);
}
//--------------------------------------------------------------------------
// child read from parent
// # Attention : it must be overridden in subclasses
//--------------------------------------------------------------------------
int CProcess::read_downstream(const char *buf)
{
	cout << buf;
	
	return 1;
}
//--------------------------------------------------------------------------
int CProcess::read_downstream(const void *buf, size_t size)
{
	for(size_t i=0;i<size;i++) cout << *((char *)buf + i);

	return 1;
}
//--------------------------------------------------------------------------
// child write to parent
//--------------------------------------------------------------------------
size_t CProcess::write_errstream(const void *buf, size_t size)
{
	return write(err_pipefd[1], buf, size);
}
//--------------------------------------------------------------------------
// child write to parent
//--------------------------------------------------------------------------
size_t CProcess::write_errstream(const char *buf)
{
	size_t size = strlen(buf);

	return write(err_pipefd[1], buf, size);
}
//--------------------------------------------------------------------------
// parent read from child 
// # Attention : it must be overridden in subclasses
//--------------------------------------------------------------------------
int CProcess::read_errstream(const char *buf)
{
	cerr << buf;

	return 1;
}
//--------------------------------------------------------------------------
int CProcess::read_errstream(const void *buf, size_t size)
{
	for(size_t i=0;i<size;i++) cerr << *((char *)buf + i);

	return 1;
}
//--------------------------------------------------------------------------
// close downstream pipe from parent to child
//--------------------------------------------------------------------------
int CProcess::close_downstream()
{
	char ch = EOF;

	write_downstream(&ch, 1);
}
//--------------------------------------------------------------------------
// close upstream pipe from child to parent
//--------------------------------------------------------------------------
int CProcess::close_upstream()
{
	char ch = EOF;

	write_upstream(&ch, 1);
}
//--------------------------------------------------------------------------
// close errstream pipe from child to parent
//--------------------------------------------------------------------------
int CProcess::close_errstream()
{
	char ch = EOF;

	write_errstream(&ch, 1);
}
//--------------------------------------------------------------------------
// 1st child processor
//--------------------------------------------------------------------------
int CProcess::input_filter()
{
	if(verbose) *err_log << "[c1]# p(" << getppid() << ") => c1(" << getpid() << ")" << endl;

	// ### use of only write downstream
	// so, no use of read/write/err upstream
	close(up_pipefd[0]);
	up_pipefd[0] = -1;
	close(up_pipefd[1]);
	up_pipefd[1] = -1;
	// also, no use of read downstream
	close(down_pipefd[0]);
	down_pipefd[0] = -1;
	// also, no use of err downstream
	close(err_pipefd[0]);
	err_pipefd[0] = -1;
	close(err_pipefd[1]);
	err_pipefd[1] = -1;

	pid = getpid(); // because pid is 0, the 1st child pid must be got
	ppid = getppid();

	// child process to achieve downstream 
	if(verbose) *err_log << "[c1]# c1(" << getpid() << ") runs before_run()" << endl;

	//--------------------------------------------------------
	before_run();

	// close downstream pipe from parent to childw 
/*
	close(down_pipefd[1]);
	down_pipefd[1] = -1;
*/
	close_downstream();
	//--------------------------------------------------------

	if(verbose) *err_log << "[c1]# c1(" << getpid() << ") is finally finished" << endl;

	return 1;
}
//--------------------------------------------------------------------------
// 2nd child process
//--------------------------------------------------------------------------
int CProcess::real_worker()
{
	if(verbose) *err_log << "[c2]# p(" << getppid() << ") => c2(" << getpid() << ")" << endl;

	// ### use of read downstream & write upstream
	// so, no use of read upstream
	close(up_pipefd[0]);
	up_pipefd[0] = -1;
	// also, no use of write downstream
	close(down_pipefd[1]);
	down_pipefd[1] = -1;
	// also, no use of read errstream
        close(err_pipefd[0]);
        err_pipefd[0] = -1;

	pid = getpid(); // because pid is 0, the 2nd pid must be got
	ppid = getppid();

	//-------------------------------------------------------------------
	// user process's stdout/stderr will be tunneled to parent process
	//-------------------------------------------------------------------
	dup2(down_pipefd[0],0); // stdin
	dup2(up_pipefd[1],1); // stdout
	dup2(err_pipefd[1],2); // stderr
	//-------------------------------------------------------------------

	if(verbose) *err_log << "[c2]# c2(" << getpid() << ") runs user process" << endl;

	//----------------------------------------------------------
	run();

	// close up/down/err stream
/*
	close(up_pipefd[1]);
	up_pipefd[1] = -1;
	close(down_pipefd[0]);
	down_pipefd[0] = -1;
	close(err_pipefd[1]);
	err_pipefd[1] = -1;
*/
	close_upstream();
	close_downstream();
	close_errstream();
	//----------------------------------------------------------

	if(verbose) *err_log << "[c2]# c2(" << getpid() << ") is finally finished" << endl;

	return 1;
}
//--------------------------------------------------------------------------
// parent process
//--------------------------------------------------------------------------
int CProcess::advisor()
{
	// ### use of only read upstream
	// so, no use of read/write downstream
	close(down_pipefd[0]);
	down_pipefd[0] = -1;
	close(down_pipefd[1]);
	down_pipefd[1] = -1;
	// also, no use of write upstream
	close(up_pipefd[1]);
	up_pipefd[1] = -1;
	// also, no use of write errstream
        close(err_pipefd[1]);
        err_pipefd[1] = -1;

	ppid = getppid();

	if(verbose) *err_log << "[p]# p(" << getpid() << ") makes c1(" << cpid[CHILD_PROC_1] << ") & c2(" << cpid[CHILD_PROC_2] << ") running" << endl;

	if(verbose) *err_log << "[p]# p(" << getpid() << ") is freezed until c1(" << cpid[CHILD_PROC_1] << ")'s finish" << endl;

	// wait for filter's finish
	filter_sync->tell_child(cpid[CHILD_PROC_1]);
	filter_sync->wait_child();

	// wait for 1st child(filter)'s finish
	//pid_t wpid = waitpid(WAIT_ANY, &filter_status, 0);
	int filter_status;
	pid_t wpid = waitpid(cpid[CHILD_PROC_1], &filter_status, 0);
/*
	int retry = 0;;
	do {
		pid_t wpid = waitpid(cpid[CHILD_PROC_1], &filter_status, WNOHANG);
		if(wpid>0) break;
		sleep(1);
	} while(retry++<5);
*/
	//assert(wpid==cpid[CHILD_PROC_1]);
}
//--------------------------------------------------------------------------
void CProcess::_start()
{
	filter_sync = new CSync();
	worker_sync = new CSync();

	// create pipes
	if(pipe(up_pipefd)<0||pipe(down_pipefd)<0||pipe(err_pipefd)<0) {
		if(verbose) *err_log << "##### p(" << getpid() << ") failed to pipe()" << endl;
		exit(-1);
	}

	pid = fork();

	if(pid==-1) {
		if(verbose) *err_log << "##### p(" << getpid() << ") failed to fork()" << endl;
		exit(-1);
	}
	// 2nd child process for user process
	else if(pid==0) { 
		if(verbose&&err_log==NULL) err_log = new CLog("/tmp/cube_proc_err.log");

		if(real_worker()) {
			// notify worker's finish
			if(verbose) *err_log << "======> [c2]notify c2[" << getpid() << "]'s finish to p(" << getppid() << ")" << endl;
			worker_sync->tell_parent(getppid());
			worker_sync->wait_parent();
			if(verbose) *err_log << "======> [c2]c2[" << getpid() << "] will be exited" << endl;
			
			exit(0);
		}
		else exit(-1);
	}
	else {
		// parent save the 2nd child process for user process
		cpid[CHILD_PROC_2] = pid; // run() itself

		int another_pid = fork();
		if(another_pid==-1) {
			if(verbose) *err_log << "##### p(" << getpid() << ") failed to another fork()" << endl;
			exit(-1);
		}
		// 1st child process for filtering input
		else if(another_pid==0) {
			if(verbose&&err_log==NULL) err_log = new CLog("/tmp/cube_proc_err.log");

			if(input_filter()) {
				// notify filter's finish
				filter_sync->tell_parent(getppid());
				filter_sync->wait_parent();

				exit(0);
			}
			else exit(-1);
		}
		// parent process for forwarding output
		else {
			// parent save the 2nd child process for user process
			cpid[CHILD_PROC_1] = another_pid;

			advisor();
		}
	}
}
//--------------------------------------------------------------------------
int CProcess::_waits()
{
	if(verbose) *err_log << "[p]# p(" << getpid() << ") is freezed until c2(" << cpid[CHILD_PROC_2] << ")'s finish" << endl;

	fd_set reads, temp_read;
	int fd_max;
	struct timeval timeout;
	int select_result;

	int size = 0;
	char buf[READ_BUF_SIZE];
	bool nomore_up_read = false;
	bool nomore_err_read = false;
	// non-blocking I/O & wait ---------------
	do {
		FD_ZERO(&reads);
		FD_SET(up_pipefd[0], &reads); // for stdout
		FD_SET(err_pipefd[0], &reads); // for stderr
		fd_max = max(up_pipefd[0], err_pipefd[0]);

		temp_read = reads;
		timeout.tv_sec = 30;
		timeout.tv_usec = 0;

		select_result = select(fd_max+1, &temp_read, (fd_set *)0, (fd_set *)0, &timeout);
		if(select_result==-1) { // select error
			// in case of massive data sort, select() might be interrupted by some signals due to slow system calls
			// so, select() must be retried
			if(errno==EINTR) {
				if(verbose) *err_log << "#### what the heck is EINTR in select()" << endl;
				exit(-1);
			}
			else {
				if(verbose) *err_log << "##### p(" << getpid() << ") failed in select() with errno = " << errno << endl;

				exit(-1);
			}
		}


		memset((char *)buf, '\0', sizeof(buf));

		if(FD_ISSET(up_pipefd[0], &temp_read)) {
			if((size=read(up_pipefd[0], buf, sizeof(buf)-1))>0) {
				buf[size] = '\0';
				if(buf[size-1]==EOF) {
					buf[size-1] = '\0';
					nomore_up_read = true;
				}
				if(buf[0]!=EOF) {
					read_upstream(buf);
				}
			}
			else {
				nomore_up_read = true;
			}
		}
		if(FD_ISSET(err_pipefd[0], &temp_read)) {
			if((size=read(err_pipefd[0], buf, sizeof(buf)-1))>0) {
				buf[size] = '\0';
				if(buf[size-1]==EOF) {
					buf[size-1] = '\0';
					nomore_err_read = true;
				}
				if(buf[0]!=EOF) {
					read_errstream(buf);
				}
			}
			else {
				nomore_err_read = true;
			}
		}
	} while(nomore_up_read==false&&nomore_err_read==false);

	// wait for worker's finish
	if(verbose) *err_log << "======> [p]wait for c2[" << cpid[CHILD_PROC_2] << "]'s finish" << endl;
	worker_sync->tell_child(cpid[CHILD_PROC_2]);
	worker_sync->wait_child();
	if(verbose) *err_log << "======> [p]notified c2[" << cpid[CHILD_PROC_2] << "]'s finish" << endl;

	// avoid child to be a zombie
	//wpid = waitpid(WAIT_ANY, &worker_status, 0);
	int worker_status;
	pid_t wpid;
	wpid = waitpid(cpid[CHILD_PROC_2], &worker_status, 0);
/*
	int retry = 0;;
	do {
		wpid = waitpid(cpid[CHILD_PROC_2], &worker_status, WNOHANG);
		if(wpid>0) break;
		sleep(1);
	} while(retry++<5);
*/
	//assert(wpid==cpid[CHILD_PROC_2]);


	if(wpid==-1) {
		if(verbose) *err_log << "##### p(" << getpid() << ") failed to wait() for c2(" << cpid[CHILD_PROC_2] << ") : errnum = " << wpid << endl;
		exit(-1);
	}
	else {
		if(WIFEXITED(worker_status)) {
			// normal exit
			if(verbose) *err_log << "[p]# p(" << getpid() << ") successfully received the exit worker_status " << WEXITSTATUS(worker_status) << " from c(" << wpid << ")" << endl;
			return WEXITSTATUS(worker_status);
		}
		else if(WIFSIGNALED(worker_status)) {
			// abort, kill or signal
		 	if(verbose) *err_log << "[p]# c(" << wpid << ") was finished with signal = " << WTERMSIG(worker_status) << (WCOREDUMP(worker_status) ? " with core file." : "") << endl;
			return -1;
		}
		else {
			if(verbose) *err_log << "[p]# c(" << wpid << ") is stopped" << endl;
			return -1;
		}
	}

	return 1;
}
//--------------------------------------------------------------------------
void CProcess::start()
{
	if(verbose) err_log = new CLog("/tmp/cube_proc_err.log");

	if(verbose) *err_log << "#### m(" << getpid() << ") first started" << endl;

	master_sync = new CSync();

	worker_pid = fork();
	if(worker_pid==-1) {
		if(verbose) *err_log << "##### m(" << getpid() << ") failed to fork()" << endl;
		exit(-1);
	}
	else if(worker_pid==0) { // slave
		if(verbose) {
			if(err_log==NULL) err_log = new CLog("/tmp/cube_proc_err.log");
			*err_log << "[p]# m(" << getppid() << ") => p(" << getpid() << ")" << endl;
		}

		_start();
		
		if(verbose) *err_log << "[p]# p(" << getpid() << ") finally started" << endl;
		int worker_status = _waits();
		if(verbose) *err_log << "[p]# p(" << getpid() << ") finally finished" << endl;

		// notify child's finish
		master_sync->tell_parent(getppid());
		master_sync->wait_parent();

		if(verbose) *err_log << "[p]# p(" << getpid() << ") will be exited with status = " << worker_status << endl;

		exit(worker_status);
	}

	// master
}
//--------------------------------------------------------------------------
int CProcess::waits()
{
	if(verbose) *err_log << "#### m(" << getpid() << ") first waiting for p(" << worker_pid << ")'s finish" << endl;

	// wait for child's finish
	master_sync->tell_child(worker_pid);
	master_sync->wait_child();

	pid_t wpid;
	//pid_t wpid = waitpid(WAIT_ANY, &master_status, 0);
	//pid_t wpid = waitpid(0, &master_status, WNOHANG);
	if(verbose) *err_log << "#### m(" << getpid() << ") second waiting for p(" << worker_pid << ")'s finish" << endl;
	wpid = waitpid(worker_pid, &status, 0);
	if(verbose) *err_log << "#### m(" << getpid() << ") notified p(" << worker_pid << ")'s finish" << endl;
/*
	int retry = 0;;
	do {
		wpid = waitpid(worker_pid, &status, WNOHANG);
		if(wpid>0) break;
		sleep(1);
	} while(retry++<5);
*/

	if(wpid==-1) {
		if(verbose) *err_log << "##### m(" << getpid() << ") failed to wait() for p(" << wpid  << ") : errnum = " << status << endl;
		exit(-1);
	}
	else {
		if(WIFEXITED(status)) {
			// normal exit
			if(verbose) *err_log << "[m]# m(" << getpid() << ") successfully received the exit status " << WEXITSTATUS(status) << " from p(" << wpid << ")" << endl;
			return WEXITSTATUS(status);
		}
		else if(WIFSIGNALED(status)) {
			// abort, kill or signal
		 	if(verbose) *err_log << "[m]# p(" << wpid << ") was finished with signal = " << WTERMSIG(status) << (WCOREDUMP(status) ? " with core file." : "") << endl;
			return -1;
		}
		else {
			if(verbose) *err_log << "[m]# p(" << wpid << ") is stopped" << endl;
			return -1;
		}
	}
}
//--------------------------------------------------------------------------
